package com.steve.bigdata.flinkmonitor.function;

import com.steve.bigdata.flinkmonitor.model.WindowResult;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Histogram;
import org.apache.flink.runtime.metrics.DescriptiveStatisticsHistogram;

/**
 * @Author: STEVE
 * @Description: 富函数实现，用于收集和上报窗口计算的监控指标
 * 继承RichMapFunction获得访问运行时上下文的能力
 * 主要功能：
 * 1. 统计处理的事件总数
 * 2. 记录数值的分布情况
 * 3. 监控助理延迟
 * @since: 2025-06-27
 */
public class MetricsMapper extends RichMapFunction<WindowResult, WindowResult> {

    // 声明为transient避免被序列化
    private transient Counter eventCounter; // 事件计数器
    private transient Histogram valueHistogram; // 数值分布直方图

    /**
     * 初始化方法，在算子启动时调用一次
     *
     * @param parameters flink配置参数
     */
    @Override
    public void open(Configuration parameters) {
        // 初始化事件计数器
        this.eventCounter = getRuntimeContext()
                .getMetricGroup()
                .counter("processedEvents");    // 指标名称会转换为metrics.reporter.prometheus中的processed_events_total
        // 初始化数值分布直方图（1000表示直方图的分桶数量）
        this.valueHistogram = getRuntimeContext()
                .getMetricGroup()
                .histogram("valueDistribution", new DescriptiveStatisticsHistogram(1000));
        // 注意：Gauge指标不需要在这里初始化，它们是懒加载的
    }

    /**
     * 处理每个窗口结果，更新监控指标
     *
     * @param result 窗口计算结果
     * @return 原样返回结果（不做修改）
     */
    @Override
    public WindowResult map(WindowResult result) {
        // 1. 更新处理事件总数（累加当前窗口的事件数）
        eventCounter.inc(result.getCount());
        // 2. 记录数值分布（将窗口总和存入直方图）
        valueHistogram.update((long) result.getSum());
        // 3. 计算并上报处理延迟（当前时间 - 窗口结束时间）
        long processingDelay = System.currentTimeMillis() - result.getWindowEnd();
        getRuntimeContext()
                .getMetricGroup()
                .gauge("processingDelayMs", () -> processingDelay); // Gauge是瞬时值，每次查询时计算
        // 透传原始结果（不做修改）
        return result;
    }

}
